package raft

import (
	"encoding/json"
	"errors"
	"fmt"
	"hash/crc32"
	"io/ioutil"
	"os"
	"path"
	"sort"
	"sync"
	"time"
)

// ---------------------------------------------------------------
// 
// 常量
// 
// --------------------------------------

const (
	Stopped      = "stopped"
	Initialized  = "initialized"
	Follower     = "follower"
	Candidate    = "candidate"
	Leader       = "leader"
	Snapshotting = "snapshotting"
)

const (
	MaxLogEntriesPerRequest         = 2000
	NumberOfLogEntriesAfterSnapshot = 200
)

const (
	// DefaultHeartbeatInterval是领导者将发送
	// 附录要求追随者保持领导力。
	DefaultHeartbeatInterval = 50 * time.Millisecond

	DefaultElectionTimeout = 150 * time.Millisecond
)

// ElectionTimeoutThresholdPercent指定服务器
// 将在其上发送心跳RTT与
// 选择超时太接近的警告事件的阈值。
const ElectionTimeoutThresholdPercent = 0.8

// ------------------------------------------------------------------
// 
// /错误
// /
// -----------------------------------------

var NotLeaderError = errors.New("raft.Server: Not current leader")
var DuplicatePeerError = errors.New("raft.Server: Duplicate peer")
var CommandTimeoutError = errors.New("raft: Command timeout")
var StopError = errors.New("raft: Has been stopped")

// -------------------------------------------------------------
// 
// Typedefs 
// 
// -------------------------------------------------------------

// /服务器参与协商一致协议，可以作为跟随者，
// 候选人或领导者。
type Server interface {
	Name() string	// 服务器名称
	Context() interface{}	// 传递给构造函数的上下文
	StateMachine() StateMachine	// 传递给构造函数的状态机。
	Leader() string	// 领导者名称
	State() string	// 服务器的当前状态。
	Path() string	// 存储路径
	LogPath() string	// 服务器的日志路径
	SnapshotPath(lastIndex uint64, lastTerm uint64) string	// 服务器的日志路径。
	Term() uint64	// 服务器的当前术语
	CommitIndex() uint64	// 服务器的当前提交索引。
	VotedFor() string	// 服务器在本学期投票给的候选人的姓名。
	MemberCount() int	// 一致性中的成员服务器数。
	QuorumSize() int	// 进行仲裁所需的服务器数。
	IsLogEmpty() bool	// 服务器日志是否没有条目
	LogEntries() []*LogEntry	// 所有日志项的列表。这只能用于调试目的。
	LastCommandName() string	// 对最后一个条目的命令名的引用。
	GetState() string	// 获取要调试的服务器状态
	ElectionTimeout() time.Duration	// 选举超时。
	SetElectionTimeout(duration time.Duration)	// 设置选择超时。
	HeartbeatInterval() time.Duration	// 检索心跳超时。
	SetHeartbeatInterval(duration time.Duration)	// 设置心跳超时。
	Transporter() Transporter	// 传输请求的对象
	SetTransporter(t Transporter)	//
	AppendEntries(req *AppendEntriesRequest) *AppendEntriesResponse	// 将零个或多个日志条目从前导添加到此服务器。
	RequestVote(req *RequestVoteRequest) *RequestVoteResponse	// 从服务器请求投票。
	RequestSnapshot(req *SnapshotRequest) *SnapshotResponse	//
	SnapshotRecoveryRequest(req *SnapshotRecoveryRequest) *SnapshotRecoveryResponse	//
	AddPeer(name string, connectiongString string) error	// 向服务器添加对等方。
	RemovePeer(name string) error	// 从服务器中删除对等方。
	Peers() map[string]*Peer	// 对等数据的副本
	Init() error	// 初始化raft服务器。
	Start() error	// 关闭服务器。
	Stop()	// 启动raft服务器
	Running() bool	// 检查服务器当前是否正在运行。
	Do(command Command) (interface{}, error)	// 尝试执行并复制命令
	TakeSnapshot() error	// 日志压缩
	LoadSnapshot() error	// 重新启动时加载快照
	AddEventListener(string, EventListener)	//
	FlushCommitIndex()	// 将提交索引刷新到磁盘。
}

type server struct {
	*eventDispatcher

	name        string
	path        string
	state       string
	transporter Transporter
	context     interface{}
	currentTerm uint64

	votedFor   string
	log        *Log
	leader     string
	peers      map[string]*Peer
	mutex      sync.RWMutex
	syncedPeer map[string]bool

	stopped           chan bool
	c                 chan *ev
	electionTimeout   time.Duration
	heartbeatInterval time.Duration

	snapshot *Snapshot

	// PendingSnapshot是一个未完成的快照。
	// 待决快照保存到磁盘后，
	// 将设置为快照，也将
	// 设置为零。
	pendingSnapshot *Snapshot

	stateMachine            StateMachine
	maxLogEntriesPerRequest uint64

	connectionString string

	routineGroup sync.WaitGroup
}

// 要由服务器的事件循环处理的内部事件。
type ev struct {
	target      interface{}
	returnValue interface{}
	c           chan error
}

// /-----------------------------------------------------------------
// 
// 构造函数
// 
// /------------------------------------------------------------------

// 在给定路径上创建带有日志的新服务器。运输商必须
// 不得为零。如果要禁用快照和日志
// 压缩，则状态机可以为零。上下文可以是任何内容（包括nil）
// 并且raft包不使用上下文，除非由
// 服务器返回。上下文（）。connectionString可以是任何东西。
func NewServer(name string, path string, transporter Transporter, stateMachine StateMachine, ctx interface{}, connectionString string) (Server, error) {
	if name == "" {
		return nil, errors.New("raft.Server: Name cannot be blank")
	}
	if transporter == nil {
		panic("raft: Transporter required")
	}

	s := &server{
		name:                    name,
		path:                    path,
		transporter:             transporter,
		stateMachine:            stateMachine,
		context:                 ctx,
		state:                   Stopped,
		peers:                   make(map[string]*Peer),
		log:                     newLog(),
		c:                       make(chan *ev, 256),
		electionTimeout:         DefaultElectionTimeout,
		heartbeatInterval:       DefaultHeartbeatInterval,
		maxLogEntriesPerRequest: MaxLogEntriesPerRequest,
		connectionString:        connectionString,
	}
	s.eventDispatcher = newEventDispatcher(s)

	// 设置应用函数。
	s.log.ApplyFunc = func(e *LogEntry, c Command) (interface{}, error) {
		// 分派提交事件。
		s.DispatchEvent(newEvent(CommitEventType, e, nil))

		// 将命令应用于状态机。
		switch c := c.(type) {
		case CommandApply:
			return c.Apply(&context{
				server:       s,
				currentTerm:  s.currentTerm,
				currentIndex: s.log.internalCurrentIndex(),
				commitIndex:  s.log.commitIndex,
			})
		case deprecatedCommandApply:
			return c.Apply(s)
		default:
			return nil, fmt.Errorf("Command does not implement Apply()")
		}
	}

	return s, nil
}

// /-----------------------------------------------------------------
// 
// /访问器
// /
// /-------------------------------------------------------------

// /------------------------------------
// 常规
// --------------------------------------

// 检索服务器的名称。
func (s *server) Name() string {
	return s.name
}

// 检索服务器的存储路径。
func (s *server) Path() string {
	return s.path
}

// 当前领导的姓名。
func (s *server) Leader() string {
	return s.leader
}

// 检索对等数据的副本。
func (s *server) Peers() map[string]*Peer {
	s.mutex.Lock()
	defer s.mutex.Unlock()

	peers := make(map[string]*Peer)
	for name, peer := range s.peers {
		peers[name] = peer.clone()
	}
	return peers
}

// 检索传输请求的对象。
func (s *server) Transporter() Transporter {
	s.mutex.RLock()
	defer s.mutex.RUnlock()
	return s.transporter
}

func (s *server) SetTransporter(t Transporter) {
	s.mutex.Lock()
	defer s.mutex.Unlock()
	s.transporter = t
}

// 检索传递给构造函数的上下文。
func (s *server) Context() interface{} {
	return s.context
}

// 检索传递给构造函数的状态机。
func (s *server) StateMachine() StateMachine {
	return s.stateMachine
}

// 检索服务器的日志路径。
func (s *server) LogPath() string {
	return path.Join(s.path, "log")
}

// 检索服务器的当前状态。
func (s *server) State() string {
	s.mutex.RLock()
	defer s.mutex.RUnlock()
	return s.state
}

// 设置服务器的状态。
func (s *server) setState(state string) {
	s.mutex.Lock()
	defer s.mutex.Unlock()

	// 临时存储以前的值。
	prevState := s.state
	prevLeader := s.leader

	// 更新状态和领导。
	s.state = state
	if state == Leader {
		s.leader = s.Name()
		s.syncedPeer = make(map[string]bool)
	}

	// 调度状态和领导变更事件。
	s.DispatchEvent(newEvent(StateChangeEventType, s.state, prevState))

	if prevLeader != s.leader {
		s.DispatchEvent(newEvent(LeaderChangeEventType, s.leader, prevLeader))
	}
}

// 检索服务器的当前术语。
func (s *server) Term() uint64 {
	s.mutex.RLock()
	defer s.mutex.RUnlock()
	return s.currentTerm
}

// 检索服务器的当前提交索引。
func (s *server) CommitIndex() uint64 {
	s.log.mutex.RLock()
	defer s.log.mutex.RUnlock()
	return s.log.commitIndex
}

// 检索此服务器在本学期投票给的候选人的姓名。
func (s *server) VotedFor() string {
	return s.votedFor
}

// 检索服务器日志是否没有条目。
func (s *server) IsLogEmpty() bool {
	return s.log.isEmpty()
}

// 所有日志项的列表。这只能用于调试目的。
func (s *server) LogEntries() []*LogEntry {
	s.log.mutex.RLock()
	defer s.log.mutex.RUnlock()
	return s.log.entries
}

// 对最后一个条目的命令名的引用。
func (s *server) LastCommandName() string {
	return s.log.lastCommandName()
}

// 获取要调试的服务器状态
func (s *server) GetState() string {
	s.mutex.RLock()
	defer s.mutex.RUnlock()
	return fmt.Sprintf("Name: %s, State: %s, Term: %v, CommitedIndex: %v ", s.name, s.state, s.currentTerm, s.log.commitIndex)
}

// 检查服务器是否可升级
func (s *server) promotable() bool {
	return s.log.currentIndex() > 0
}

// /--------------------------------------------
// 成员
// /---------------------------------------

// 检索一致性中的成员服务器数。
func (s *server) MemberCount() int {
	s.mutex.RLock()
	defer s.mutex.RUnlock()
	return len(s.peers) + 1
}

// 检索进行仲裁所需的服务器数。
func (s *server) QuorumSize() int {
	return (s.MemberCount() / 2) + 1
}

// /---------------------------------------
// 选举超时
// /---------------------------------------

// 检索选举超时。
func (s *server) ElectionTimeout() time.Duration {
	s.mutex.RLock()
	defer s.mutex.RUnlock()
	return s.electionTimeout
}

// 设置选择超时。
func (s *server) SetElectionTimeout(duration time.Duration) {
	s.mutex.Lock()
	defer s.mutex.Unlock()
	s.electionTimeout = duration
}

// /---------------------------------------
// 心跳超时
// /---------------------------------------

// 检索心跳超时。
func (s *server) HeartbeatInterval() time.Duration {
	s.mutex.RLock()
	defer s.mutex.RUnlock()
	return s.heartbeatInterval
}

// 设置心跳超时。
func (s *server) SetHeartbeatInterval(duration time.Duration) {
	s.mutex.Lock()
	defer s.mutex.Unlock()

	s.heartbeatInterval = duration
	for _, peer := range s.peers {
		peer.setHeartbeatInterval(duration)
	}
}

// -------------------------------------------------------------
// 
// 方法
// 
// -------------------------------------------------------------

// ------------------------------------
// 初始化
// ------------------------------------

// 注册NOP命令
func init() {
	RegisterCommand(&NOPCommand{})
	RegisterCommand(&DefaultJoinCommand{})
	RegisterCommand(&DefaultLeaveCommand{})
}

// 启动raft服务器
// 如果存在日志条目，则在未收到AEs的情况下允许升级到候选服务器。
// 如果不存在日志项，则等待来自其他节点的AEs。
// 如果不存在日志条目，并且发出了自连接命令，则
// 立即成为leader并提交条目。
func (s *server) Start() error {
	// 如果服务器已在运行，请退出。
	if s.Running() {
		return fmt.Errorf("raft.Server: Server already running[%v]", s.state)
	}

	if err := s.Init(); err != nil {
		return err
	}

	// 每次服务器启动
	// 时都需要分配stopped，因为它在“Stop”处关闭。
	s.stopped = make(chan bool)
	s.setState(Follower)

	// 如果不存在日志项，则
	// 1。等待来自另一个节点
	// 2的AEs。等待self-join命令
	// 将其自身设置为可升级
	if !s.promotable() {	// 服务器是否可升级
		s.debugln("start as a new raft server")

		// 如果存在日志项，则允许升级到候选
		// 如果未收到AEs。
	} else {
		s.debugln("start from previous saved state")
	}

	debugln(s.GetState()) // 获取要调试的服务器状态

	s.routineGroup.Add(1)
	go func() {
		defer s.routineGroup.Done()
		s.loop()	// 函数将等待
	}()

	return nil
}

// 初始化raft服务器。
// 如果给定路径下没有以前的日志文件，Init（）将创建一个空日志文件。
// 否则，Init（）将从日志文件加载日志项。
func (s *server) Init() error {
	if s.Running() {
		return fmt.Errorf("raft.Server: Server already running[%v]", s.state)
	}

	// 服务器已初始化或初始化后服务器已停止
	// 如果日志已初始化，则我们知道服务器已在
	// 运行后停止。
	if s.state == Initialized || s.log.initialized {
		s.state = Initialized
		return nil
	}

	// 如果快照目录不存在，则创建快照目录
	err := os.Mkdir(path.Join(s.path, "snapshot"), 0700)
	if err != nil && !os.IsExist(err) {
		s.debugln("raft: Snapshot dir error: ", err)
		return fmt.Errorf("raft: Initialization error: %s", err)
	}

	if err := s.readConf(); err != nil {
		s.debugln("raft: Conf file error: ", err)
		return fmt.Errorf("raft: Initialization error: %s", err)
	}

	// 初始化日志并将其加载。
	if err := s.log.open(s.LogPath()); err != nil {
		s.debugln("raft: Log error: ", err)
		return fmt.Errorf("raft: Initialization error: %s", err)
	}

	// 将术语更新为日志中的最后一个术语。
	_, s.currentTerm = s.log.lastInfo()

	s.state = Initialized
	return nil
}

// 关闭服务器。
func (s *server) Stop() {
	if s.State() == Stopped {
		return
	}

	close(s.stopped)

	// 在关闭日志之前，确保所有goroutine都已停止
	s.routineGroup.Wait()

	s.log.close()
	s.setState(Stopped)
}

// 检查服务器当前是否正在运行。
func (s *server) Running() bool {
	s.mutex.RLock()
	defer s.mutex.RUnlock()
	return (s.state != Stopped && s.state != Initialized)
}

// /---------------------------------------
// 术语
// /---------------------------------------

// 更新服务器的当前术语。仅当找到更大的
// 外部术语时才使用。
func (s *server) updateCurrentTerm(term uint64, leaderName string) {
	_assert(term > s.currentTerm,
		"upadteCurrentTerm: update is called when term is not larger than currentTerm")

	// 暂时存储以前的值。
	prevTerm := s.currentTerm
	prevLeader := s.leader

	// 设置currentTerm=T，转换为follower（§5.1）
	// 在降压前停止心跳
	if s.state == Leader {
		for _, peer := range s.peers {
			peer.stopHeartbeat(false)
		}
	}
	// 更新
	if s.state != Follower {
		s.setState(Follower)
	}

	s.mutex.Lock()
	s.currentTerm = term
	s.leader = leaderName
	s.votedFor = ""
	s.mutex.Unlock()

	s.DispatchEvent(newEvent(TermChangeEventType, s.currentTerm, prevTerm))

	if prevLeader != s.leader {
		s.DispatchEvent(newEvent(LeaderChangeEventType, s.leader, prevLeader))
	}
}


// 函数将等待
func (s *server) loop() {
	defer s.debugln("server.loop.end")

	state := s.State()

	for state != Stopped {
		s.debugln("server.loop.run ", state)
		switch state {
		case Follower:
			s.followerLoop()
		case Candidate:
			s.candidateLoop()
		case Leader:
			s.leaderLoop()
		case Snapshotting:
			s.snapshotLoop()
		}
		state = s.State()
	}
}

// 直到事件实际处理完毕，然后返回。
func (s *server) send(value interface{}) (interface{}, error) {
	if !s.Running() {
		return nil, StopError
	}

	event := &ev{target: value, c: make(chan error, 1)}
	select {
	case s.c <- event:
	case <-s.stopped:
		return nil, StopError
	}
	select {
	case <-s.stopped:
		return nil, StopError
	case err := <-event.c:
		return event.returnValue, err
	}
}

func (s *server) sendAsync(value interface{}) {
	if !s.Running() {
		return
	}

	event := &ev{target: value, c: make(chan error, 1)}
	// 首先尝试非阻塞发送
	// 在大多数情况下，这不应阻塞
	// 避免创建不必要的go例程
	select {
	case s.c <- event:
		return
	default:
	}

	s.routineGroup.Add(1)
	go func() {
		defer s.routineGroup.Done()
		select {
		case s.c <- event:
		case <-s.stopped:
		}
	}()
}

// 服务器处于跟随者状态时运行的事件循环。
// 回应候选人和领导人的RPC。
// 如果选举超时，则转换为候选人，但没有：
// 1。接收有效的附加条目RPC或
// 2。授予候选人投票权
func (s *server) followerLoop() {
	since := time.Now()
	electionTimeout := s.ElectionTimeout()
	timeoutChan := afterBetween(s.ElectionTimeout(), s.ElectionTimeout()*2)

	for s.State() == Follower {
		var err error
		update := false
		select {
		case <-s.stopped:
			s.setState(Stopped)
			return

		case e := <-s.c:
			switch req := e.target.(type) {
			case JoinCommand:
				// 如果不存在日志条目，并且发出了自加入命令
				// 则立即成为领导者并提交条目。
				if s.log.currentIndex() == 0 && req.NodeName() == s.Name() {
					s.debugln("selfjoin and promote to leader")
					s.setState(Leader)
					s.processCommand(req, e)
				} else {
					err = NotLeaderError
				}
			case *AppendEntriesRequest:
				// 如果心跳太接近选举超时，则发送事件。
				elapsedTime := time.Now().Sub(since)
				if elapsedTime > time.Duration(float64(electionTimeout)*ElectionTimeoutThresholdPercent) {
					s.DispatchEvent(newEvent(ElectionTimeoutThresholdEventType, elapsedTime, nil))
				}
				e.returnValue, update = s.processAppendEntriesRequest(req)
			case *RequestVoteRequest:
				e.returnValue, update = s.processRequestVoteRequest(req)
			case *SnapshotRequest:
				e.returnValue = s.processSnapshotRequest(req)
			default:
				err = NotLeaderError
			}
			// 事件的回调。
			e.c <- err

		case <-timeoutChan:
			// 只允许同步追随者提升为候选
			if s.promotable() {
				s.setState(Candidate)
			} else {
				update = true
			}
		}

		// 如果选举超时，则转换为候选人，但不带：
		// 1。接收有效的附加条目RPC或
		// 2。授予候选
		if update {
			since = time.Now()
			timeoutChan = afterBetween(s.ElectionTimeout(), s.ElectionTimeout()*2)
		}
	}
}

// 当服务器处于候选状态时运行的事件循环。
func (s *server) candidateLoop() {
	// 清除前导值。
	prevLeader := s.leader
	s.leader = ""
	if prevLeader != s.leader {
		s.DispatchEvent(newEvent(LeaderChangeEventType, s.leader, prevLeader))
	}

	lastLogIndex, lastLogTerm := s.log.lastInfo()
	doVote := true
	votesGranted := 0
	var timeoutChan <-chan time.Time
	var respChan chan *RequestVoteResponse

	for s.State() == Candidate {
		if doVote {
			// 增加当前任期，为自己投票。
			s.currentTerm++
			s.votedFor = s.name

			// 将RequestVote RPC发送到所有其他服务器。
			respChan = make(chan *RequestVoteResponse, len(s.peers))
			for _, peer := range s.peers {
				s.routineGroup.Add(1)
				go func(peer *Peer) {
					defer s.routineGroup.Done()
					peer.sendVoteRequest(newRequestVoteRequest(s.currentTerm, s.name, lastLogIndex, lastLogTerm), respChan)
				}(peer)
			}

			// 等待：
			// /*从大多数服务器收到的投票：成为领导者
			// /*从新领导者收到的附加条目：下台。
			// /*选举超时，无选举决议：增加任期，开始新选举
			// /*发现更高任期：下台（§5.1）
			votesGranted = 1
			timeoutChan = afterBetween(s.ElectionTimeout(), s.ElectionTimeout()*2)
			doVote = false
		}

		// 如果我们获得足够的选票，请停止等待更多选票。
		// 并从候选人循环返回
		if votesGranted == s.QuorumSize() {
			s.debugln("server.candidate.recv.enough.votes")
			s.setState(Leader)
			return
		}

		// 从同行收集选票。
		select {
		case <-s.stopped:
			s.setState(Stopped)
			return

		case resp := <-respChan:
			if success := s.processVoteResponse(resp); success {
				s.debugln("server.candidate.vote.granted: ", votesGranted)
				votesGranted++
			}

		case e := <-s.c:
			var err error
			switch req := e.target.(type) {
			case Command:
				err = NotLeaderError
			case *AppendEntriesRequest:
				e.returnValue, _ = s.processAppendEntriesRequest(req)
			case *RequestVoteRequest:
				e.returnValue, _ = s.processRequestVoteRequest(req)
			}

			// 事件的回调。
			e.c <- err

		case <-timeoutChan:
			doVote = true
		}
	}
}

// 服务器处于引导状态时运行的事件循环。
func (s *server) leaderLoop() {
	logIndex, _ := s.log.lastInfo()

	// 将对等方的PrevLoginEx更新为leader的LastLoginEx并启动heartbeat。
	s.debugln("leaderLoop.set.PrevIndex to ", logIndex)
	for _, peer := range s.peers {
		peer.setPrevLogIndex(logIndex)
		peer.startHeartbeat()
	}

	// 在服务器成为领导者后提交NOP。从草稿纸：
	// “选举时：向每个服务器发送初始空附件条目RPC（心跳）；在空闲期间重复以防止选举超时
	// （§5.2）”。上面开始的心跳做“空闲”期间的工作。
	s.routineGroup.Add(1)
	go func() {
		defer s.routineGroup.Done()
		s.Do(NOPCommand{})
	}()

	// 开始收集追随者的响应
	for s.State() == Leader {
		var err error
		select {
		case <-s.stopped:
			// 停止前停止所有对等方
			for _, peer := range s.peers {
				peer.stopHeartbeat(false)
			}
			s.setState(Stopped)
			return

		case e := <-s.c:
			switch req := e.target.(type) {
			case Command:
				s.processCommand(req, e)
				continue
			case *AppendEntriesRequest:
				e.returnValue, _ = s.processAppendEntriesRequest(req)
			case *AppendEntriesResponse:
				s.processAppendEntriesResponse(req)
			case *RequestVoteRequest:
				e.returnValue, _ = s.processRequestVoteRequest(req)
			}

			// 回调事件。
			e.c <- err
		}
	}

	s.syncedPeer = nil
}

func (s *server) snapshotLoop() {
	for s.State() == Snapshotting {
		var err error
		select {
		case <-s.stopped:
			s.setState(Stopped)
			return

		case e := <-s.c:
			switch req := e.target.(type) {
			case Command:
				err = NotLeaderError
			case *AppendEntriesRequest:
				e.returnValue, _ = s.processAppendEntriesRequest(req)
			case *RequestVoteRequest:
				e.returnValue, _ = s.processRequestVoteRequest(req)
			case *SnapshotRecoveryRequest:
				e.returnValue = s.processSnapshotRecoveryRequest(req)
			}
			// 事件的回调。
			e.c <- err
		}
	}
}

// /---------------------------------------
// 命令
// /---------------------------------------

// 尝试执行并复制命令。当成功提交命令或发生错误时，函数将返回
// 。

func (s *server) Do(command Command) (interface{}, error) {
	return s.send(command)
}

// 处理一个命令。
func (s *server) processCommand(command Command, e *ev) {
	s.debugln("server.command.process")

	// 在日志中为命令创建一个条目。
	entry, err := s.log.createEntry(s.currentTerm, command, e)

	if err != nil {
		s.debugln("server.command.log.entry.error:", err)
		e.c <- err
		return
	}

	if err := s.log.appendEntry(entry); err != nil {
		s.debugln("server.command.log.error:", err)
		e.c <- err
		return
	}

	s.syncedPeer[s.Name()] = true
	if len(s.peers) == 0 {
		commitIndex := s.log.currentIndex()
		s.log.setCommitIndex(commitIndex)
		s.debugln("commit index ", commitIndex)
	}
}

// /---------------------------------------
// 追加条目
// /---------------------------------------

// 将零个或多个日志条目从前导添加到此服务器。
func (s *server) AppendEntries(req *AppendEntriesRequest) *AppendEntriesResponse {
	ret, _ := s.send(req)
	resp, _ := ret.(*AppendEntriesResponse)
	return resp
}

// 处理“追加条目”请求。
func (s *server) processAppendEntriesRequest(req *AppendEntriesRequest) (*AppendEntriesResponse, bool) {
	s.traceln("server.ae.process")

	if req.Term < s.currentTerm {
		s.debugln("server.ae.error: stale term")
		return newAppendEntriesResponse(s.currentTerm, false, s.log.currentIndex(), s.log.CommitIndex()), false
	}

	if req.Term == s.currentTerm {
		_assert(s.State() != Leader, "leader.elected.at.same.term.%d\n", s.currentTerm)

		// 当它是一个候选者
		if s.state == Candidate {
			// 将状态更改为follower 
			s.setState(Follower)
		}

		// 当候选者
		// 当follower 
		s.leader = req.LeaderName
	} else {
		// 更新期限和领导时发现新领导。
		s.updateCurrentTerm(req.Term, req.LeaderName)
	}

	// 如果日志不包含匹配的前一项，则拒绝。
	if err := s.log.truncate(req.PrevLogIndex, req.PrevLogTerm); err != nil {
		s.debugln("server.ae.truncate.error: ", err)
		return newAppendEntriesResponse(s.currentTerm, false, s.log.currentIndex(), s.log.CommitIndex()), true
	}

	// 将条目追加到日志中。
	if err := s.log.appendEntries(req.Entries); err != nil {
		s.debugln("server.ae.append.error: ", err)
		return newAppendEntriesResponse(s.currentTerm, false, s.log.currentIndex(), s.log.CommitIndex()), true
	}

	// 提交到提交索引。
	if err := s.log.setCommitIndex(req.CommitIndex); err != nil {
		s.debugln("server.ae.commit.error: ", err)
		return newAppendEntriesResponse(s.currentTerm, false, s.log.currentIndex(), s.log.CommitIndex()), true
	}

	// 一旦服务器追加并提交了来自领导者的所有日志条目

	return newAppendEntriesResponse(s.currentTerm, true, s.log.currentIndex(), s.log.CommitIndex()), true
}

// 就会处理来自对等方的“追加条目”响应。这只是
// 当服务器是领导者时处理的。在其他
// 状态期间收到的响应将被丢弃。
func (s *server) processAppendEntriesResponse(resp *AppendEntriesResponse) {
	// 如果我们找到一个更高的术语，那么就换成跟随者并退出。
	if resp.Term() > s.Term() {
		s.updateCurrentTerm(resp.Term(), "")
		return
	}

	// 如果失败，会出现恐慌反应。
	if !resp.Success() {
		return
	}

	// 如果一个对等方成功地从前导项附加日志，
	// 我们将其添加到同步列表
	if resp.append == true {
		s.syncedPeer[resp.peer] = true
	}

	// 增加提交计数，以确保在提交之前有一个仲裁。
	if len(s.syncedPeer) < s.QuorumSize() {
		return
	}

	// 确定多数人已提交的索引。
	var indices []uint64
	indices = append(indices, s.log.currentIndex())
	for _, peer := range s.peers {
		indices = append(indices, peer.getPrevLogIndex())
	}
	sort.Sort(sort.Reverse(uint64Slice(indices)))

	// 我们最多可以提交大多数成员附加的索引。
	commitIndex := indices[s.QuorumSize()-1]
	committedIndex := s.log.commitIndex

	if commitIndex > committedIndex {
		// leader在提交日志项之前需要进行fsync 
		s.log.sync()
		s.log.setCommitIndex(commitIndex)
		s.debugln("commit index ", commitIndex)
	}
}

// ProcessVotereResponse处理投票请求：
// 1。如果候选人的当前任期获得投票权，请返回真实的
// 2。如果由于任期较小而导致投票被拒绝，请更新此服务器
// 的任期，这也将导致候选人下台，并返回false。
// 3。如果投票是为了一个较小的任期，忽略它并返回false。
func (s *server) processVoteResponse(resp *RequestVoteResponse) bool {
	if resp.VoteGranted && resp.Term == s.currentTerm {
		return true
	}

	if resp.Term > s.currentTerm {
		s.debugln("server.candidate.vote.failed")
		s.updateCurrentTerm(resp.Term, "")
	} else {
		s.debugln("server.candidate.vote: denied")
	}
	return false
}

// /---------------------------------------
// 请求投票
// /---------------------------------------

// 从服务器请求投票。如果服务器当前任期内的投票期限为
// 且服务器尚未进行投票，则可以获得投票。如果期限大于服务器的当前期限，也可以获得
// 的投票。
func (s *server) RequestVote(req *RequestVoteRequest) *RequestVoteResponse {
	ret, _ := s.send(req)
	resp, _ := ret.(*RequestVoteResponse)
	return resp
}

// 处理“请求投票”请求。
func (s *server) processRequestVoteRequest(req *RequestVoteRequest) (*RequestVoteResponse, bool) {

	// 如果请求来自旧术语，则拒绝该请求。
	if req.Term < s.Term() {
		s.debugln("server.rv.deny.vote: cause stale term")
		return newRequestVoteResponse(s.currentTerm, false), false
	}

	// 如果请求对等方的术语大于此节点，则更新术语
	// 如果任期相等，我们已经投票给另一位候选人，那么
	// 不要投票给这位候选人。
	if req.Term > s.Term() {
		s.updateCurrentTerm(req.Term, "")
	} else if s.votedFor != "" && s.votedFor != req.CandidateName {
		s.debugln("server.deny.vote: cause duplicate vote: ", req.CandidateName,
			" already vote for ", s.votedFor)
		return newRequestVoteResponse(s.currentTerm, false), false
	}

	// 如果候选人的日志至少不如我们上次的日志最新，那么不要投票。
	lastIndex, lastTerm := s.log.lastInfo()
	if lastIndex > req.LastLogIndex || lastTerm > req.LastLogTerm {
		s.debugln("server.deny.vote: cause out of date log: ", req.CandidateName,
			"Index :[", lastIndex, "]", " [", req.LastLogIndex, "]",
			"Term :[", lastTerm, "]", " [", req.LastLogTerm, "]")
		return newRequestVoteResponse(s.currentTerm, false), false
	}

	// 如果我们能做到这一点，那么就投票并重新设定我们的选举时间。
	s.debugln("server.rv.vote: ", s.name, " votes for", req.CandidateName, "at term", req.Term)
	s.votedFor = req.CandidateName

	return newRequestVoteResponse(s.currentTerm, true), true
}

// /---------------------------------------
// 成员身份
// /---------------------------------------

// 向服务器添加对等方。
func (s *server) AddPeer(name string, connectiongString string) error {
	s.debugln("server.peer.add: ", name, len(s.peers))

	// 不允许添加两次对等点。
	if s.peers[name] != nil {
		return nil
	}

	// 如果对等方与服务器同名，则跳过该对等方
	if s.name != name {
		peer := newPeer(s, name, connectiongString, s.heartbeatInterval)

		if s.State() == Leader {
			peer.startHeartbeat()
		}

		s.peers[peer.Name] = peer

		s.DispatchEvent(newEvent(AddPeerEventType, name, nil))
	}

	// 将配置写入文件。
	s.writeConf()

	return nil
}

// 从服务器中删除对等方。
func (s *server) RemovePeer(name string) error {
	s.debugln("server.peer.remove: ", name, len(s.peers))

	// 如果对等方与服务器同名，则跳过该对等方
	if name != s.Name() {
		// 如果对等方不存在，则返回错误。
		peer := s.peers[name]
		if peer == nil {
			return fmt.Errorf("raft: Peer not found: %s", name)
		}

		// 停止对等并将其删除。
		if s.State() == Leader {
			// 我们在这里创建一个go例程以避免潜在的死锁。
			// 当到达这行代码时，我们将保持日志写锁。
			// 对等。如果
			// 目标go例程（我们要停止）正在调用
			// 日志，则可以在不使用go例程的情况下阻止stopHeartbeat。getEntriesAfter并等待日志读取锁定。
			// 所以我们可能持有日志锁并等待日志锁，
			// 这会导致死锁。
			// TODO（向里）重构日志锁
			s.routineGroup.Add(1)
			go func() {
				defer s.routineGroup.Done()
				peer.stopHeartbeat(true)
			}()
		}

		delete(s.peers, name)

		s.DispatchEvent(newEvent(RemovePeerEventType, name, nil))
	}

	// 将配置写入文件。
	s.writeConf()

	return nil
}

// /---------------------------------------
// 日志压缩
// /---------------------------------------

func (s *server) TakeSnapshot() error {
	if s.stateMachine == nil {
		return errors.New("Snapshot: Cannot create snapshot. Missing state machine.")
	}

	// 无锁快捷方式
	// 如果服务器当前正在创建快照，请退出。
	if s.pendingSnapshot != nil {
		return errors.New("Snapshot: Last snapshot is not finished.")
	}

	// TODO:获取锁，不允许再提交
	// 这将在完成重构心跳后完成
	s.debugln("take.snapshot")

	lastIndex, lastTerm := s.log.commitInfo()

	// 检查自
	// 上次快照以来是否提交了日志。
	if lastIndex == s.log.startIndex {
		return nil
	}

	path := s.SnapshotPath(lastIndex, lastTerm)
	// 将快照附加到挂起的快照并将其保存到磁盘。
	s.pendingSnapshot = &Snapshot{lastIndex, lastTerm, nil, nil, path}

	state, err := s.stateMachine.Save()
	if err != nil {
		return err
	}

	// 克隆对等点列表。
	peers := make([]*Peer, 0, len(s.peers)+1)
	for _, peer := range s.peers {
		peers = append(peers, peer.clone())
	}
	peers = append(peers, &Peer{Name: s.Name(), ConnectionString: s.connectionString})

	// 将快照附加到挂起的快照并将其保存到磁盘。
	s.pendingSnapshot.Peers = peers
	s.pendingSnapshot.State = state
	s.saveSnapshot()

	// 我们在快照之后保留了一些日志条目。
	// 我们不希望将整个快照发送到速度稍慢的机器
	if lastIndex-s.log.startIndex > NumberOfLogEntriesAfterSnapshot {
		compactIndex := lastIndex - NumberOfLogEntriesAfterSnapshot
		compactTerm := s.log.getEntry(compactIndex).Term()
		s.log.compact(compactIndex, compactTerm)
	}

	return nil
}

// 检索服务器的日志路径。
func (s *server) saveSnapshot() error {
	if s.pendingSnapshot == nil {
		return errors.New("pendingSnapshot.is.nil")
	}

	// 将快照写入磁盘。
	if err := s.pendingSnapshot.save(); err != nil {
		return err
	}

	// 交换当前和最后的快照。
	tmp := s.snapshot
	s.snapshot = s.pendingSnapshot

	// 如果有任何更改，请删除以前的快照
	if tmp != nil && !(tmp.LastIndex == s.snapshot.LastIndex && tmp.LastTerm == s.snapshot.LastTerm) {
		tmp.remove()
	}
	s.pendingSnapshot = nil

	return nil
}

// 检索服务器的日志路径。
func (s *server) SnapshotPath(lastIndex uint64, lastTerm uint64) string {
	return path.Join(s.path, "snapshot", fmt.Sprintf("%v_%v.ss", lastTerm, lastIndex))
}

func (s *server) RequestSnapshot(req *SnapshotRequest) *SnapshotResponse {
	ret, _ := s.send(req)
	resp, _ := ret.(*SnapshotResponse)
	return resp
}

func (s *server) processSnapshotRequest(req *SnapshotRequest) *SnapshotResponse {
	// 如果跟随者的日志在快照的最后一个索引中包含一个条目，该条目的词条
	// 与快照的最后一个词条匹配，则跟随者已经拥有快照中找到的所有
	// 信息，并且可以回答false。
	entry := s.log.getEntry(req.LastIndex)

	if entry != nil && entry.Term() == req.LastTerm {
		return newSnapshotResponse(false)
	}

	// 更新状态。
	s.setState(Snapshotting)

	return newSnapshotResponse(true)
}

func (s *server) SnapshotRecoveryRequest(req *SnapshotRecoveryRequest) *SnapshotRecoveryResponse {
	ret, _ := s.send(req)
	resp, _ := ret.(*SnapshotRecoveryResponse)
	return resp
}

func (s *server) processSnapshotRecoveryRequest(req *SnapshotRecoveryRequest) *SnapshotRecoveryResponse {
	// 恢复从请求发送的状态。
	if err := s.stateMachine.Recovery(req.State); err != nil {
		panic("cannot recover from previous state")
	}

	// 恢复群集配置。
	s.peers = make(map[string]*Peer)
	for _, peer := range req.Peers {
		s.AddPeer(peer.Name, peer.ConnectionString)
	}

	// 更新日志状态。
	s.currentTerm = req.LastTerm
	s.log.updateCommitIndex(req.LastIndex)

	// 创建本地快照。
	s.pendingSnapshot = &Snapshot{req.LastIndex, req.LastTerm, req.Peers, req.State, s.SnapshotPath(req.LastIndex, req.LastTerm)}
	s.saveSnapshot()

	// 清除以前的日志条目。
	s.log.compact(req.LastIndex, req.LastTerm)

	return newSnapshotRecoveryResponse(req.LastTerm, true, req.LastIndex)
}

// 重新启动时加载快照
func (s *server) LoadSnapshot() error {
	// 打开快照/目录。
	dir, err := os.OpenFile(path.Join(s.path, "snapshot"), os.O_RDONLY, 0)
	if err != nil {
		s.debugln("cannot.open.snapshot: ", err)
		return err
	}

	// 检索所有快照的列表。
	filenames, err := dir.Readdirnames(-1)
	if err != nil {
		dir.Close()
		panic(err)
	}
	dir.Close()

	if len(filenames) == 0 {
		s.debugln("no.snapshot.to.load")
		return nil
	}

	// 抓取最新快照。
	sort.Strings(filenames)
	snapshotPath := path.Join(s.path, "snapshot", filenames[len(filenames)-1])

	// 读取快照数据。
	file, err := os.OpenFile(snapshotPath, os.O_RDONLY, 0)
	if err != nil {
		return err
	}
	defer file.Close()

	// 检查校验和。
	var checksum uint32
	n, err := fmt.Fscanf(file, "%08x\n", &checksum)
	if err != nil {
		return err
	} else if n != 1 {
		return errors.New("checksum.err: bad.snapshot.file")
	}

	// 加载剩余快照内容。
	b, err := ioutil.ReadAll(file)
	if err != nil {
		return err
	}

	// 生成校验和。
	byteChecksum := crc32.ChecksumIEEE(b)
	if uint32(checksum) != byteChecksum {
		s.debugln(checksum, " ", byteChecksum)
		return errors.New("bad snapshot file")
	}

	// 解码快照。
	if err = json.Unmarshal(b, &s.snapshot); err != nil {
		s.debugln("unmarshal.snapshot.error: ", err)
		return err
	}

	// 将快照恢复到状态机。
	if err = s.stateMachine.Recovery(s.snapshot.State); err != nil {
		s.debugln("recovery.snapshot.error: ", err)
		return err
	}

	// 恢复群集配置。
	for _, peer := range s.snapshot.Peers {
		s.AddPeer(peer.Name, peer.ConnectionString)
	}

	// 更新日志状态。
	s.log.startTerm = s.snapshot.LastTerm
	s.log.startIndex = s.snapshot.LastIndex
	s.log.updateCommitIndex(s.snapshot.LastIndex)

	return err
}

// /---------------------------------------
// 配置文件
// /---------------------------------------

// 将提交索引刷新到磁盘。
// 因此，当raft服务器重新启动时，它将提交到刷新的commitIndex。
func (s *server) FlushCommitIndex() {
	s.debugln("server.conf.update")
	// 将配置写入文件。
	s.writeConf()
}

func (s *server) writeConf() {

	peers := make([]*Peer, len(s.peers))

	i := 0
	for _, peer := range s.peers {
		peers[i] = peer.clone()
		i++
	}

	r := &Config{
		CommitIndex: s.log.commitIndex,
		Peers:       peers,
	}

	b, _ := json.Marshal(r)

	confPath := path.Join(s.path, "conf")
	tmpConfPath := path.Join(s.path, "conf.tmp")

	err := writeFileSynced(tmpConfPath, b, 0600)

	if err != nil {
		panic(err)
	}

	os.Rename(tmpConfPath, confPath)
}

// 读取服务器的配置。
func (s *server) readConf() error {
	confPath := path.Join(s.path, "conf")
	s.debugln("readConf.open ", confPath)

	// 打开conf文件
	b, err := ioutil.ReadFile(confPath)

	if err != nil {
		return nil
	}

	conf := &Config{}

	if err = json.Unmarshal(b, conf); err != nil {
		return err
	}

	s.log.updateCommitIndex(conf.CommitIndex)

	return nil
}

// /-----------------------------------------
// 调试
// /------------------------------------

func (s *server) debugln(v ...interface{}) {
	if logLevel > Debug {
		debugf("[%s Term:%d] %s", s.name, s.Term(), fmt.Sprintln(v...))
	}
}

func (s *server) traceln(v ...interface{}) {
	if logLevel > Trace {
		tracef("[%s] %s", s.name, fmt.Sprintln(v...))
	}
}
